/*
 * Example_06.cpp
 *
 *  Created on: Mar 15, 2009
 *      Author: Lev
 */

#include <iostream>
#include <exception>
#include <string>
#include <sstream>
#include <cstring>
#include <queue>
#include <ctime>
#include <pthread.h>
#include <errno.h>
using namespace std;

// Pthread exception - includes status code from errno.h
class pthread_error : public exception {
private:
	string _message;
public:
	pthread_error( const char *msg = "", int status = 0 ) : _message("pthread_error: ") {
		ostringstream oss( ostringstream::out );
		if( strlen( msg ) > 0 )
			oss << msg;
		else
			oss << "unknown";
		switch( status ) {
		case EINVAL:
			oss << " (EINVAL)";
			break;
		case EPERM:
			oss << " (EPERM)";
			break;
		case EBUSY:
			oss << " (EBUSY)";
			break;
		case EAGAIN:
			oss << " (EAGAIN)";
			break;
		case EDEADLK:
			oss << " (EDEADLK)";
			break;
		default:
			break;
		}
		_message += oss.str();
	}
	const char* what() { return _message.c_str(); }
	~pthread_error() throw() {}
};

// sample queue - starts in separate thread
template<typename T>
class SampleQueue {
private:
	queue<T> _queue;
	timespec _timeout;
	pthread_mutex_t _mutex;
	pthread_cond_t _cond_var;
	bool _terminated;
	// checks internal queue
	bool isEmpty() const {
		return _queue.empty();
	}
	// removes and processes front element from internal queue
	void processFrontElement() {
		T &element = _queue.front();
		_queue.pop();
		ostringstream oss( ostringstream::out );
		oss << "SampleQueue processing: " << element << endl;
		cout << oss.str() << flush; // using oss to make atomic output
	}
public:
	// starts queue processing thread
	// static function pointer is used in pthread_create
	static void* start_queue( void *arg ) {
		SampleQueue<T> *queue = (SampleQueue<T> *)arg;
		try {
			queue->process();
		} catch( pthread_error &error ) {
			ostringstream oss( ostringstream::out );
			oss << "SampleQueue::start error: " << error.what() << endl;
			cout << oss.str() << flush;
		}
		return NULL;
	}
	// initializes mutex and cond_var
	SampleQueue() : _terminated(false) {
		_mutex = PTHREAD_MUTEX_INITIALIZER;
		_cond_var = PTHREAD_COND_INITIALIZER;
	}
	// adds element to the queue end and signals cond_var
	void pushElement( const T &element ) throw(pthread_error) {
		int status = pthread_mutex_lock( &_mutex ); // enter mutex
		if( status != 0 )
			throw pthread_error( "pushElement - mutex lock failed", status );
		_queue.push( element );
		status = pthread_mutex_unlock( &_mutex ); // leave mutex
		if( status != 0 )
			throw pthread_error( "pushElement - mutex unlock failed", status );
		status = pthread_cond_signal( &_cond_var ); // signal cond_var
		if( status != 0 )
			throw pthread_error( "pushElement - cond signal failed", status );
	}
	// waits on cond_var until signaled, then processes front elements
	void process() throw(pthread_error) {
		ostringstream oss1( ostringstream::out );
		oss1 << "SampleQueue: started" << endl;
		cout << oss1.str() << flush;
		while( true ) {
			_timeout.tv_sec = time(NULL) + 1; // timeout = 1 sec
			_timeout.tv_nsec = 0;
			int status = pthread_mutex_lock( &_mutex ); // enter mutex
			if( status != 0 )
				throw pthread_error( "process - mutex lock failed", status );
			bool terminated = _terminated; // copy member flag to local var
			if( ! terminated ) {
				try {
					if( isEmpty() ) {
						// wait timeout on cond_var
						status = pthread_cond_timedwait( &_cond_var, &_mutex, &_timeout );
						if( status != 0 && status != ETIMEDOUT )
							throw pthread_error( "process - cond timedwait failed", status );
					}
					while( ! isEmpty() ) // process front element
						processFrontElement();
				} catch( pthread_error &error ) {
					oss1.seekp( ios::beg );
					oss1 << error.what() << endl;
					cout << oss1.str() << flush;
				}
			}
			status = pthread_mutex_unlock( &_mutex ); // leave mutex
			if( status != 0 )
				throw pthread_error( "process - mutex unlock failed", status );
			if( terminated )
				break;
		}
		ostringstream oss2( ostringstream::out );
		oss2 << "SampleQueue: terminated" << endl;
		cout << oss2.str() << flush;
	}
	// sets termination flag to stop queue processing thread and signals cond_var
	void terminate() throw(pthread_error) {
		int status = pthread_mutex_lock( &_mutex ); // enter mutex
		if( status != 0 )
			throw pthread_error( "terminate - mutex lock failed", status );
		_terminated = true;
		status = pthread_mutex_unlock( &_mutex ); // leave mutex
		if( status != 0 )
			throw pthread_error( "terminate - mutex unlock failed", status );
		status = pthread_cond_signal( &_cond_var ); // signal cond_var
		if( status != 0 )
			throw pthread_error( "terminate - cond signal failed", status );
	}
	// destroys mutex and cond_var
	~SampleQueue() {
		pthread_mutex_destroy( &_mutex );
		pthread_cond_destroy( &_cond_var );
	}
};

//void* start_queue( void *arg ) {
//	SampleQueue<double> *queue = (SampleQueue<double> *)arg;
//	try {
//		queue->process();
//	} catch( pthread_error &error ) {
//		ostringstream oss( ostringstream::out );
//		oss << "start_queue error: " << error.what() << endl;
//		cout << oss.str();
//	}
//	return NULL;
//}

void* startThreadTest( void *arg ) {
	SampleQueue<double> *queue = (SampleQueue<double> *)arg;
	// start thread iterations
	ostringstream oss( ostringstream::out );
	oss << "startThreadTest: starting ..." << endl;
	cout << oss.str() << flush;
	for( int i = 100; i < 110; i++ ) {
		sleep( 1 );
		try {
			queue->pushElement( (double)i );
		} catch( pthread_error &error ) {
			ostringstream oss1( ostringstream::out );
			oss1 << "startThreadTest error: " << error.what() << endl;
			cout << oss1.str() << flush;
		}
	}
	return NULL;
}

int __main() {
	const int N = 10; // number of iterations
	pthread_t queueThreadId, testThreadId;
	void *threadResult;
	int status;

	// create queue object and static start_queue function pointer
	SampleQueue<double> queue;
	void* (*ptStartQueue)(void *) = &SampleQueue<double>::start_queue;
	ostringstream oss( ostringstream::out );
	oss << "main: starting queue ... " << endl;
	cout << oss.str() << flush;
	// start queue processing thread
	status = pthread_create( &queueThreadId, NULL, ptStartQueue, &queue );
	if( status != 0 ) {
		pthread_error error( "pthread_create failed", status );
		ostringstream oss1( ostringstream::out );
		oss1 << error.what() << endl;
		cout << oss1.str() << flush;
		return status;
	}
	// start separate thread test
	status = pthread_create( &testThreadId, NULL, startThreadTest, &queue );
	if( status != 0 ) {
		pthread_error error( "pthread_create failed", status );
		ostringstream oss1( ostringstream::out );
		oss1 << error.what() << endl;
		cout << oss1.str() << flush;
		return status;
	}
	// start main thread iterations
	for( int i = 0; i < N; i++ ) {
		sleep( 1 );
		try {
			queue.pushElement( (double)(i + 1) );
		} catch( pthread_error &error ) {
			ostringstream oss1( ostringstream::out );
			oss1 << "main error: " << error.what() << endl;
			cout << oss1.str() << flush;
		}
	}
	sleep( 1 );
	// wait until test thread finishes
	status = pthread_join( testThreadId, &threadResult );
	if( status != 0 ) {
		pthread_error error( "pthread_join failed", status );
		ostringstream oss1( ostringstream::out );
		oss1 << error.what() << endl;
		cout << oss1.str() << flush;
		return status;
	}
	// terminating queue processing thread
	try {
		oss.seekp( ios::beg );
		oss << "main: terminating queue ... " << endl;
		cout << oss.str() << flush;
		queue.terminate();
	} catch( pthread_error &error ) {
		oss.seekp( ios::beg );
		oss << "main error: " << error.what() << endl;
		cout << oss.str() << flush;
	}
	// wait until queue thread finishes
	status = pthread_join( queueThreadId, &threadResult );
	if( status != 0 ) {
		pthread_error error( "pthread_join failed", status );
		ostringstream oss1( ostringstream::out );
		oss1 << error.what() << endl;
		cout << oss1.str() << flush;
		return status;
	}
	return 0;
}
